package com.lvmama.rhino.analyze.nginx

import com.lvmama.rhino.common.utils.JDBCUtil.JDBCTemplate
import com.lvmama.rhino.common.utils.Utils
import org.apache.spark.sql.SaveMode
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.types.StringType

/**
  * Created by yuanxiaofeng on 2016/6/1.
  */
object ManGodPageFlow {
  private val AppName = "ManGodPageFlowJob"
  def execute(path: String, jars: Seq[String] = Nil ): Unit = {
    val sc = {
      val conf = new SparkConf().setAppName(AppName).setJars(jars)
      new SparkContext(conf)
    }
    val hiveCtx = new HiveContext(sc)
    import  hiveCtx.implicits._
    val ngDS= sc.textFile(path).map(l => Utils.initNginx(l)).toDS()
    ngDS.persist(StorageLevel.MEMORY_ONLY_SER)

    val JDBCDefault = JDBCTemplate.JDBCDefaultSet
    val connP = JDBCTemplate.getConProperties

    val filterNgDs = ngDS.filter(_.logType == "forward")
    val pre = filterNgDs.map(ng => (ng.pageCode, ng.refer)).toDF().coalesce(4).groupBy($"_1", $"_2").agg(count("*") as "pre")
      .withColumn("pages",  concat_ws(":", col("_2"), col("pre"))).groupBy(col("_1")).agg(collect_list(col("pages")).as("tmp"))
      .withColumn("supPage", col("tmp").cast(StringType)).drop(col("tmp"))
    val next = filterNgDs.map(ng => (ng.refer, ng.pageCode)).toDF().coalesce(4).groupBy($"_1", $"_2").agg(count("*") as "next")
      .withColumn("pages",  concat_ws(":", col("_2"), col("next"))).groupBy(col("_1")).agg(collect_list(col("pages")).as("tmp"))
      .withColumn("subPage", col("tmp").cast(StringType)).drop(col("tmp"))
    val relation = pre.join(next, "_1").withColumnRenamed("_1", "curPage")
    relation.select("curPage", "supPage", "subPage").write.mode(SaveMode.Append).option("driver", JDBCDefault.get("driver").get)
      .jdbc(JDBCDefault.get("conn").get, "relationship", connP)

    val eventStat = ngDS.filter(_.logType == "click").map(ng => (ng.pageCode, ng.eventId)).toDF().coalesce(4).groupBy(col("_1")
      .as("page_code"), col("_2").as("event_id")).agg(count("*").as("click_count")).withColumn("project_name", lit("mangod").cast(StringType))
    eventStat.select("page_code", "event_id", "click_count", "project_name").write.mode(SaveMode.Append).option("driver", JDBCDefault.get("driver").get)
      .jdbc(JDBCDefault.get("conn").get, "event_click_count", connP)

    sc.stop()
  }
}
